// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#include <nebula/array/array_binary.h>

#include <cstdint>
#include <memory>

#include <nebula/array/array_base.h>
#include <nebula/array/validate.h>
#include <nebula/types/type.h>
#include <nebula/types/type_traits.h>
#include <nebula/util/binary_view_util.h>
#include <turbo/base/checked_cast.h>
#include <turbo/log/logging.h>

namespace nebula {

    BinaryArray::BinaryArray(const std::shared_ptr<ArrayData> &data) {
        KCHECK(is_binary_like(data->type->id()));
        set_data(data);
    }

    BinaryArray::BinaryArray(int64_t length, const std::shared_ptr<Buffer> &value_offsets,
                             const std::shared_ptr<Buffer> &data,
                             const std::shared_ptr<Buffer> &null_bitmap, int64_t null_count,
                             int64_t offset) {
        set_data(ArrayData::create(binary(), length, {null_bitmap, value_offsets, data},
                                null_count, offset));
    }

    LargeBinaryArray::LargeBinaryArray(const std::shared_ptr<ArrayData> &data) {
        KCHECK(is_large_binary_like(data->type->id()));
        set_data(data);
    }

    LargeBinaryArray::LargeBinaryArray(int64_t length,
                                       const std::shared_ptr<Buffer> &value_offsets,
                                       const std::shared_ptr<Buffer> &data,
                                       const std::shared_ptr<Buffer> &null_bitmap,
                                       int64_t null_count, int64_t offset) {
        set_data(ArrayData::create(large_binary(), length, {null_bitmap, value_offsets, data},
                                null_count, offset));
    }

    StringArray::StringArray(const std::shared_ptr<ArrayData> &data) {
        KCHECK_EQ(data->type->id(), Type::STRING);
        set_data(data);
    }

    StringArray::StringArray(int64_t length, const std::shared_ptr<Buffer> &value_offsets,
                             const std::shared_ptr<Buffer> &data,
                             const std::shared_ptr<Buffer> &null_bitmap, int64_t null_count,
                             int64_t offset) {
        set_data(ArrayData::create(utf8(), length, {null_bitmap, value_offsets, data}, null_count,
                                offset));
    }

    turbo::Status StringArray::validate_utf8() const { return internal::validate_utf8(*data_); }

    LargeStringArray::LargeStringArray(const std::shared_ptr<ArrayData> &data) {
        KCHECK_EQ(data->type->id(), Type::LARGE_STRING);
        set_data(data);
    }

    LargeStringArray::LargeStringArray(int64_t length,
                                       const std::shared_ptr<Buffer> &value_offsets,
                                       const std::shared_ptr<Buffer> &data,
                                       const std::shared_ptr<Buffer> &null_bitmap,
                                       int64_t null_count, int64_t offset) {
        set_data(ArrayData::create(large_utf8(), length, {null_bitmap, value_offsets, data},
                                null_count, offset));
    }

    turbo::Status LargeStringArray::validate_utf8() const { return internal::validate_utf8(*data_); }

    BinaryViewArray::BinaryViewArray(std::shared_ptr<ArrayData> data) {
        KCHECK_EQ(data->type->id(), Type::BINARY_VIEW);
        set_data(std::move(data));
    }

    BinaryViewArray::BinaryViewArray(std::shared_ptr<DataType> type, int64_t length,
                                     std::shared_ptr<Buffer> views, BufferVector buffers,
                                     std::shared_ptr<Buffer> null_bitmap, int64_t null_count,
                                     int64_t offset) {
        buffers.insert(buffers.begin(), std::move(views));
        buffers.insert(buffers.begin(), std::move(null_bitmap));
        set_data(
                ArrayData::create(std::move(type), length, std::move(buffers), null_count, offset));
    }

    std::string_view BinaryViewArray::get_view(int64_t i) const {
        const std::shared_ptr<Buffer> *data_buffers = data_->buffers.data() + 2;
        return util::FromBinaryView(raw_values_[i], data_buffers);
    }

    StringViewArray::StringViewArray(std::shared_ptr<ArrayData> data) {
        KCHECK_EQ(data->type->id(), Type::STRING_VIEW);
        set_data(std::move(data));
    }

    turbo::Status StringViewArray::validate_utf8() const { return internal::validate_utf8(*data_); }

    FixedSizeBinaryArray::FixedSizeBinaryArray(const std::shared_ptr<ArrayData> &data) {
        set_data(data);
    }

    FixedSizeBinaryArray::FixedSizeBinaryArray(const std::shared_ptr<DataType> &type,
                                               int64_t length,
                                               const std::shared_ptr<Buffer> &data,
                                               const std::shared_ptr<Buffer> &null_bitmap,
                                               int64_t null_count, int64_t offset)
            : PrimitiveArray(type, length, data, null_bitmap, null_count, offset),
              byte_width_(turbo::checked_cast<const FixedSizeBinaryType &>(*type).byte_width()) {}

    const uint8_t *FixedSizeBinaryArray::get_value(int64_t i) const {
        return raw_values_ + (i + data_->offset) * byte_width_;
    }

}  // namespace nebula
